package com.im;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.im.pojo.*;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.RedisTemplate;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class ClientSample {

    @Autowired
    private RedisTemplate redisTemplate;

    private static HashMap<String, Object> message = new HashMap<>();

    @Test
    public void main() {
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("1.13.16.98", 11111), "example", "", "");
        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("等待");
                } else {
                    System.out.println(message);
//                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }

        } finally {
            connector.disconnect();
        }
    }

    public static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }

            //entry.getHeader().getLogfileName();//binlog文件
            //entry.getHeader().getLogfileOffset();//binlog偏移量

            EventType eventType = rowChage.getEventType();
            String schemaName = entry.getHeader().getSchemaName();//数据库
            String tableName = entry.getHeader().getTableName();//数据表

            if (schemaName.equals("im") && eventType != EventType.QUERY) {
                JSONObject json = getJson(rowChage, eventType);
                switch (tableName) {
                    case "im_message":
                        ImMessage imMessage = json.toJavaObject(ImMessage.class);
                        message.put(imMessage.getMessage_id(), imMessage);
//                        redisTemplate.opsForValue().set(imMessage.getMessage_id(), imMessage);
                        break;
                    case "im_lately":
                        ImLately imLately = json.toJavaObject(ImLately.class);
                        HashMap<String, String> hashMap = new HashMap<>();
                        hashMap.put("send_id", imLately.getSend_id());
                        hashMap.put("receive_id", imLately.getReceive_id());
                        hashMap.put("message_id", imLately.getMessage_id());
                        hashMap.put("read_message", String.valueOf(imLately.getRead_message()));
//                        redisTemplate.opsForHash().putAll(imLately.getSend_id(), hashMap);
                        break;
                    case "im_send":
                        ImSend imSend = json.toJavaObject(ImSend.class);
                        break;
                    case "im_user":
                        ImUser imUser = json.toJavaObject(ImUser.class);
                        break;
                    case "im_apply":
                        ImApply imApply = json.toJavaObject(ImApply.class);
                        break;
                    default:
                        break;
                }
            }

        }
    }

    private static JSONObject getJson(RowChange rowChage, EventType eventType) {

        JSONObject jsonObject = new JSONObject();

        for (RowData r : rowChage.getRowDatasList()) {
            if (eventType == EventType.DELETE) {
                for (Column c : r.getBeforeColumnsList()) {
                    jsonObject.put(c.getName(), c.getValue());
                }
            } else {
                for (Column c : r.getAfterColumnsList()) {
                    jsonObject.put(c.getName(), c.getValue());
                }
            }
        }

        return jsonObject;
    }
}
